package com.example.flinkcourse.lesson5.sink;

import com.example.flinkcourse.lesson5.config.Config;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Doris Sink 实现
 */
public class DorisSink {
    private static final Logger LOG = LoggerFactory.getLogger(DorisSink.class);

    /**
     * 创建 Doris Sink
     *
     * @param config 配置信息
     * @param tableName 表名
     * @return SinkFunction
     */
    public static SinkFunction<String> createSink(Config config, String tableName) {
        LOG.info("创建 Doris Sink, 表名: {}", tableName);
        
        // 配置 Doris 连接信息
        DorisOptions.Builder dorisBuilder = DorisOptions.builder()
            .setFenodes(config.getDorisHost() + ":" + config.getDorisPort())
            .setTableIdentifier(config.getDorisDatabase() + "." + tableName)
            .setUsername(config.getDorisUsername())
            .setPassword(config.getDorisPassword());
        
        // 配置 Doris 执行选项
        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");
        
        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder()
            .setLabelPrefix("doris_sink")  // 导入标签前缀
            .setStreamLoadProp(properties)  // Stream Load 属性
            .setDeletable(false);  // 是否支持删除
        
        // 配置 Doris 读取选项
        DorisReadOptions.Builder readBuilder = DorisReadOptions.builder()
            .setDeserializeArrowAsync(false)
            .setDeserializeQueueSize(64)
            .setExecMemLimit(2147483648L)
            .setRequestQueryTimeoutS(3600)
            .setRequestBatchSize(1000)
            .setRequestConnectTimeoutMs(10000)
            .setRequestReadTimeoutMs(10000)
            .setRequestRetries(3)
            .setRequestTabletSize(1024 * 1024);
        
        // 创建 Doris Sink
        return DorisSink.<String>builder()
            .setDorisOptions(dorisBuilder.build())
            .setDorisReadOptions(readBuilder.build())
            .setDorisExecutionOptions(executionBuilder.build())
            .setSerializer((element, context) -> element.getBytes())  // 序列化器
            .build();
    }
} 